/******************************************************************************
 * Filename : uc_prcom.c
 * Copyright:
 * Created  : 2014-4-11 by franco.yin
 * Description -
 *
 ******************************************************************************/

#ifdef __cplusplus
	 extern "C" {
#endif

#include "uc_prcom_var.h"

ucprcom_func ucprcom_cli_function = NULL;
UC_U32 ucprcom_cli_flag = 0;
UC_U32 ucprcom_cli_status = 0;
void *ucprcom_cli_void_p = NULL;

extern int pthread_mutex_consistent_np(ucprcom_lock_s *mutex);

void
ucprcom_sleep(int ms)
{
	struct timeval delay_time;
	int ret = 0;

	if (ms <= 0) return;

	delay_time.tv_sec = ms / 1000;
	delay_time.tv_usec = (ms % 1000)*1000;
	while(1) {
		ret = select(0, NULL, NULL, NULL, &delay_time);
		if(ret < 0 && errno == EINTR) {
			continue;
		}
		break;
	}
	return;
}

void
ucprcom_uptime(ucprcom_timestamp_s *pTimestamp, ucprcom_timestamp_s *pLastTimestamp, int fps)
{
	struct timespec ts;
	int64_t time_tmp = 0;
	int64_t time_tmp_div = 0;
	int64_t time_tmp_last = 0;
	int64_t time_tmp_next = 0;
	int64_t fps_inv = 0;

	clock_gettime(CLOCK_MONOTONIC, &ts);

	if(fps <= 0) {
		pTimestamp->sec = ts.tv_sec;
		pTimestamp->usec = ts.tv_nsec / 1000;
	} else {
		fps_inv = 1000000 / fps;
		time_tmp = (((int64_t)ts.tv_sec) * 1000000 + ts.tv_nsec / 1000);
		time_tmp_div = (time_tmp / fps_inv) * fps_inv;
		if(pLastTimestamp == NULL) {
			if(time_tmp - time_tmp_div >= (fps_inv / 2)) {
				time_tmp_div += fps_inv;
			}
		} else {
			time_tmp_last = (((int64_t)pLastTimestamp->sec) * 1000000 + pLastTimestamp->usec);
			if(time_tmp_last > 0) {
				time_tmp_next = time_tmp_last + fps_inv;
				if((time_tmp_div <= time_tmp_last) || (time_tmp_div > time_tmp_next && time_tmp_div <= time_tmp_next + fps_inv + fps_inv)) {
					time_tmp_div = time_tmp_next;
				}
			}
		}
		pTimestamp->sec = time_tmp_div / 1000000;
		pTimestamp->usec = time_tmp_div - (pTimestamp->sec * 1000000);
	}

	if(pLastTimestamp != NULL) {
		memcpy(pLastTimestamp, pTimestamp, sizeof(ucprcom_timestamp_s));
	}
}

int
ucprcom_system(char *pFmt, ...)
{
	va_list ap;
	char cmd[10240];

	va_start(ap, pFmt);
	vsnprintf(cmd, sizeof(cmd), pFmt, ap);
	va_end(ap);

	//uc_p("ucprcom_system[%s]\n", cmd);

	return system(cmd);
}

int
ucprcom_system_ret(char *result, int result_len, char *pFmt, ...)
{
    FILE *stream = NULL;
	va_list ap;
	char cmd[10240];
	int n = 0;

	result[0] = '\0';
	va_start(ap, pFmt);
	vsnprintf(cmd, sizeof(cmd), pFmt, ap);
	va_end(ap);

	//uc_p("ucprcom_system_ret[%s]\n", cmd);

    stream = popen(cmd, "r");
	if(stream == NULL) {
		uc_err("popen error while[%s]\n", cmd);
		return -1;
	}
    n = fread(result, 1, result_len - 1, stream);
	result[n] = '\0';

    pclose(stream);
    return 0;
}

int
ucprcom_uptime_diff_ms(ucprcom_timestamp_s *pStartTime, ucprcom_timestamp_s *pEndTime)
{
	return ((pEndTime->sec - pStartTime->sec) * 1000 +  pEndTime->usec / 1000) - (pStartTime->usec / 1000) ;
}

int
ucprcom_lock_init(ucprcom_lock_s *pLock)
{
	pthread_mutexattr_t attr;
	if (pthread_mutexattr_init(&attr) != 0
		|| pthread_mutexattr_setpshared(&attr, PTHREAD_PROCESS_SHARED) != 0
		|| pthread_mutexattr_setrobust_np(&attr, PTHREAD_MUTEX_ROBUST_NP) != 0) {
		ucprcom_p("ucprcom: lock init error[%d:%s]\n", errno, strerror(errno));
		return -1;
	}

	pthread_mutex_init(pLock, &attr);

	pthread_mutexattr_destroy(&attr);

	return 0;
}

int
ucprcom_lock(ucprcom_lock_s *pLock)
{
	if (pthread_mutex_lock(pLock) == EOWNERDEAD) {
		if (pthread_mutex_consistent_np(pLock) != 0) {
			ucprcom_p("ucprcom: lock consistent_np error[%d:%s]\n", errno, strerror(errno));
			pthread_mutex_unlock(pLock);
			return -1;
		} else {
			ucprcom_p("ucprcom: lock reset while owner not exist\n");
		}
	}

	return 0;
}

int
ucprcom_unlock(ucprcom_lock_s *pLock)
{
	return pthread_mutex_unlock(pLock);
}

void
ucprcom_sig_init(ucprcom_sig_s *pSig)
{
	pthread_cond_init(&(pSig->cond), NULL);
	pthread_mutex_init(&(pSig->mutex), NULL);
}

void
ucprcom_sig_wait(ucprcom_sig_s *pSig)
{
	pthread_mutex_lock(&(pSig->mutex));
	pthread_cond_wait(&(pSig->cond), &(pSig->mutex));
	pthread_mutex_unlock(&(pSig->mutex));
}

int
ucprcom_sig_wait_timeout(ucprcom_sig_s *pSig, int msec)
{
	struct timespec ts;
	struct timeval now_time;
	int ret = 0;

	gettimeofday(&now_time, NULL);
	ts.tv_sec = now_time.tv_sec + msec / 1000;
	ts.tv_nsec = now_time.tv_usec * 1000 + (msec % 1000) * 1000000;
	if (ts.tv_nsec >= 1000000000) {
		ts.tv_sec += 1;
		ts.tv_nsec -= 1000000000;
	}

	pthread_mutex_lock(&(pSig->mutex));
	if(msec > 0) {
		ret = pthread_cond_timedwait(&(pSig->cond), &(pSig->mutex), (const struct timespec *)&ts);
	} else {
		ret = pthread_cond_wait(&(pSig->cond), &(pSig->mutex));
	}
	pthread_mutex_unlock(&(pSig->mutex));

	return ret;
}

void
ucprcom_sig_destroy(ucprcom_sig_s *pSig)
{
	pthread_cond_destroy(&(pSig->cond));
	pthread_mutex_destroy(&(pSig->mutex));
}

int
ucprcom_trim_to(char *dst, char *src, int max_dst_len)
{
	int i = 0;
	int len = 0;
	int head = -1;
	int tail = -1;

	if(dst != NULL && dst != src) {
		*dst = 0;
	}

	if(dst == NULL || src == NULL) {
		return -1;
	}

	len = strlen(src);
	for(i = 0; i < len; i++) {
		if(src[i] != ' ' && src[i] != '\r' && src[i] != '\n' && src[i] != '\t') {
			head = i;
			break;
		}
	}

	if(head >= 0) {
		for(i = len - 1; i >= 0; i--) {
			if(src[i] != ' ' && src[i] != '\r' && src[i] != '\n' && src[i] != '\t') {
				tail = i;
				break;
			}
		}
		len = tail - head + 1;
		if(max_dst_len > 0) {
			len = uc_min(len, max_dst_len - 1);
		}
		if(len > 0) {
			memmove(dst, src + head, len);
			dst[len] = 0;
			return len;
		} else {
			return 0;
		}
	} else {
		return 0;
	}
}

int
ucprcom_get_cfg(char *file, char *key, char *value, int value_len)
{
	char key_tmp[2048];
	char line[2048];
	char *pTmp = NULL;
	FILE *fp = NULL;
	int line_len = 0;
	int key_len = 0;
	int ret = -1;

	if(value != NULL) {
		*value = 0;
	} else {
		ucprcom_p("ucprcom_get_cfg bad entry\n");
		return -1;
	}

	if(file == NULL || key == NULL) {
		ucprcom_p("ucprcom_get_cfg bad entry\n");
		return -1;
	}

	if ((fp = fopen(file, "r")) == NULL) {
		return -1;
	}

	ucprcom_trim_to(key_tmp, key, sizeof(key_tmp));
	key_len = strlen(key_tmp);

	while (fgets(line, sizeof(line), fp) != NULL) {
		line_len = ucprcom_trim_to(line, line, -1);
		if(line_len < key_len
			|| line[0] == '#'
			|| ((pTmp = strstr(line, key_tmp)) == NULL)
			|| pTmp != line
			|| (*(pTmp + key_len) != 0 && strchr(UCCFG_KEY_DIVIDE_RANGE, *(pTmp + key_len)) == NULL)) {
			continue;
		}
		ret = 0;

		pTmp += key_len;
		if(*pTmp != 0) {
			pTmp++;
			while (*pTmp != 0 && strchr(UCCFG_KEY_DIVIDE_RANGE, *pTmp) != NULL) {
				pTmp++;
			}
			if(*pTmp != 0) {
				strncpy(value, pTmp, value_len);
			}
		}
		break;
	}

	fclose(fp);

	return ret;
}

int
ucprcom_add_cli(ucprcom_shrm_s *pShrm, int pid, UC_U32 flag)
{
	int i = 0;

	for(i = 0; i < UCPRCOM_SHRM_CLI_MAX; i++) {
		if(pid == pShrm->ctrl.arr_pid[i] || pShrm->ctrl.arr_pid[i] <= 0) {
			if(pShrm->ctrl.arr_pid[i] <= 0) {
				ucprcom_p("ucprcom: ++ add[%d] pid[%d] flag[0x%08X]\n", i, pid, flag);
			} else {
				ucprcom_p("ucprcom: UU update[%d] pid[%d] flag[0x%08X]\n", i, pid, flag);
			}
			pShrm->ctrl.arr_pid[i] = pid;
			pShrm->ctrl.arr_flag[i] = flag;
			return 0;
		}
	}

	ucprcom_p("-------------\n");

	for(i = 0; i < UCPRCOM_SHRM_CLI_MAX; i++) {
		ucprcom_p("[%d] pid[%d] flag[0x%08X]\n", i, pShrm->ctrl.arr_pid[i], pShrm->ctrl.arr_flag[i]);
	}

	ucprcom_p("ucprcom: pid[%d] flag[0x%08X] shrm cli full\n", pid, flag);
	return -1;
}

int
ucprcom_del_cli_by_pid(ucprcom_shrm_s *pShrm, int pid)
{
	int i = 0;

	for(i = 0; i < UCPRCOM_SHRM_CLI_MAX; i++) {
		if(pid == pShrm->ctrl.arr_pid[i]) {
			ucprcom_p("ucprcom: -- del[%d] pid[%d] flag[0x%08X]\n", i, pShrm->ctrl.arr_pid[i], pShrm->ctrl.arr_flag[i]);
			pShrm->ctrl.arr_pid[i] = 0;
			pShrm->ctrl.arr_flag[i] = 0;
		}
	}

	return 0;
}

int
ucprcom_del_cli_by_index(ucprcom_shrm_s *pShrm, int i)
{
	if(i < 0 || i >= UCPRCOM_SHRM_CLI_MAX) {
		return -1;
	}

	ucprcom_p("ucprcom: -- del[%d] pid[%d] flag[0x%08X]\n", i, pShrm->ctrl.arr_pid[i], pShrm->ctrl.arr_flag[i]);
	pShrm->ctrl.arr_pid[i] = 0;
	pShrm->ctrl.arr_flag[i] = 0;

	return 0;
}

int
ucprcom_cli_get_data(ucprcom_data_head_s *pHead, UC_U8 *pData, ucprcom_timestamp_s *pTimestamp)
{
	int ret = 0;

	if(ucprcom_lock(&(pHead->lock)) != 0) {
		return -1;
	}

	if(ucprcom_cli_function != NULL
		&& ucprcom_get_bit(pHead->flag, UCPRCOM_DATA_FALG_OUTPUT) > 0
		&& (pTimestamp->sec < pHead->creat_time.sec
			|| (pTimestamp->sec == pHead->creat_time.sec
				&& pTimestamp->usec <= pHead->creat_time.usec))) {
		ucprcom_cli_function(pHead->type, &(pHead->fmt), pData, pHead->len, ucprcom_cli_void_p);
		memcpy(pTimestamp, &pHead->creat_time, sizeof(ucprcom_timestamp_s));
		ret = 1;
	}

	ucprcom_unlock(&(pHead->lock));

	return ret;
}

int
ucprcom_cli_get_v(ucprcom_shrm_s *pShrm, int *pIndex, ucprcom_timestamp_s *pTimestamp)
{
	int ret = ucprcom_cli_get_data(&(pShrm->arr_v[*pIndex].head), pShrm->arr_v[*pIndex].data, pTimestamp);

	if(ret > 0) {
		*pIndex = *pIndex + 1;
		if(*pIndex >= UCPRCOM_SHRM_CACHE) {
			*pIndex = 0;
		}
	}

	return ret;
}

int
ucprcom_cli_get_a(ucprcom_shrm_s *pShrm, int *pIndex, ucprcom_timestamp_s *pTimestamp)
{
	int ret = ucprcom_cli_get_data(&(pShrm->arr_a[*pIndex].head), pShrm->arr_a[*pIndex].data, pTimestamp);

	if(ret > 0) {
		*pIndex = *pIndex + 1;
		if(*pIndex >= UCPRCOM_SHRM_CACHE) {
			*pIndex = 0;
		}
	}

	return ret;
}

int
ucprcom_cli_get_ts(ucprcom_shrm_s *pShrm, int *pIndex, ucprcom_timestamp_s *pTimestamp)
{
	int ret = ucprcom_cli_get_data(&(pShrm->arr_ts[*pIndex].head), pShrm->arr_ts[*pIndex].data, pTimestamp);

	if(ret > 0) {
		*pIndex = *pIndex + 1;
		if(*pIndex >= UCPRCOM_SHRM_CACHE) {
			*pIndex = 0;
		}
	}

	return ret;
}

int
ucprcom_cli_get_bmp(ucprcom_shrm_s *pShrm, int *pIndex, ucprcom_timestamp_s *pTimestamp)
{
	int ret = ucprcom_cli_get_data(&(pShrm->arr_pic[*pIndex].head), pShrm->arr_pic[*pIndex].data, pTimestamp);

	if(ret > 0) {
		*pIndex = *pIndex + 1;
		if(*pIndex >= UCPRCOM_SHRM_CACHE) {
			*pIndex = 0;
		}
	}

	return ret;
}

int
ucprcom_cli_start(uint32_t flag, ucprcom_func func, void *p)
{
	if(func == NULL) {
		ucprcom_p("ucprcom: func NULL\n");
		return -1;
	}

	ucprcom_cli_flag = flag;
	ucprcom_cli_function = func;
	ucprcom_cli_void_p = p;
	ucprcom_cli_status = 1;

	return 0;
}

int
ucprcom_cli_stop()
{
	ucprcom_cli_status = 0;
	return 0;
}

void
ucprcom_cli_get_input(int *pInpsel, int *pVfmt, int *pWidth, int *pHeight, int *pFps, int *pInterlace)
{
	char buf[128];

	if(ucprcom_get_cfg(UC_CFG_FILE_VFMT, UC_CFG_KEY_INPSEL, buf, sizeof(buf)) == 0) {
		if(pInpsel != NULL) *pInpsel = atoi(buf);
	} else {
		if(pInpsel != NULL) *pInpsel = -1;
	}
	if(ucprcom_get_cfg(UC_CFG_FILE_VFMT, UC_CFG_KEY_VFMT, buf, sizeof(buf)) == 0) {
		if(pVfmt != NULL) *pVfmt = atoi(buf);
	} else {
		if(pVfmt != NULL) *pVfmt = -1;
	}
	if(ucprcom_get_cfg(UC_CFG_FILE_VFMT, UC_CFG_KEY_INTERLACE, buf, sizeof(buf)) == 0) {
		if(pInterlace != NULL) *pInterlace = atoi(buf);
	} else {
		if(pInterlace != NULL) *pInterlace = 0;
	}
	if(ucprcom_get_cfg(UC_CFG_FILE_VFMT, UC_CFG_KEY_FPS, buf, sizeof(buf)) == 0) {
		if(pFps != NULL) *pFps = atoi(buf);
	} else {
		if(pFps != NULL) *pFps = 0;
	}
	if(ucprcom_get_cfg(UC_CFG_FILE_VFMT, UC_CFG_KEY_WIDTH, buf, sizeof(buf)) == 0) {
		if(pWidth != NULL) *pWidth = atoi(buf);
	} else {
		if(pWidth != NULL) *pWidth = 0;
	}
	if(ucprcom_get_cfg(UC_CFG_FILE_VFMT, UC_CFG_KEY_HEIGHT, buf, sizeof(buf)) == 0) {
		if(pHeight != NULL) *pHeight = atoi(buf);
	} else {
		if(pHeight != NULL) *pHeight = 0;
	}
}

void *
ucprcom_cli_thread(void *p)
{
	ucprcom_shrm_s *pShrm = NULL;
	int shmid;
	int index_v = 0;
	int index_a = 0;
	int index_ts = 0;
	int index_bmp = 0;
	int running_status = 0;
	ucprcom_timestamp_s timestamp_v;
	ucprcom_timestamp_s timestamp_a;
	ucprcom_timestamp_s timestamp_ts;
	ucprcom_timestamp_s timestamp_bmp;
	int suspend = 0;
	int pid = getpid();
	int ret = 0;
	int fifo_fd = -1;
	char fifo_file[512];
	char fifo_buf[512];

	sprintf(fifo_file, "%sucprcom_%d", UCPRCOM_FIFO_PATH, pid);
	while(1) {
		ucprcom_system("rm -f %s", fifo_file);
		if (mkfifo(fifo_file, 0777) == 0) {
			break;
		}
		ucprcom_p("ucprcom: creat fifo error[%d:%s]\n", errno, strerror(errno));
		ucprcom_sleep(3000);
	}

	while(1) {

		shmid = shmget(UCPRCOM_SHRM_KEY, sizeof(ucprcom_shrm_s), 0666);
		if (shmid < 0) {
			ucprcom_p("ucprcom: connect shrm error[%d:%s]\n", errno, strerror(errno));
			ucprcom_sleep(3000);
			continue;
		}

		pShrm = (ucprcom_shrm_s *)shmat(shmid, NULL, 0);
		if(pShrm == NULL) {
			ucprcom_p("ucprcom: shmat error[%d:%s]\n", errno, strerror(errno));
			ucprcom_sleep(3000);
			continue;
		}

		ucprcom_lock(&(pShrm->ctrl.lock));
		index_v = pShrm->ctrl.index_v;
		index_a = pShrm->ctrl.index_a;
		index_ts = pShrm->ctrl.index_ts;
		index_bmp = pShrm->ctrl.index_bmp;
		ucprcom_p("ucprcom: cli start with[%d][%d][%d][%d]\n", index_v, index_a, index_ts, index_bmp);
		memset(&timestamp_v, 0, sizeof(ucprcom_timestamp_s));
		memset(&timestamp_a, 0, sizeof(ucprcom_timestamp_s));
		memset(&timestamp_ts, 0, sizeof(ucprcom_timestamp_s));
		if(ucprcom_add_cli(pShrm, pid, ucprcom_cli_flag) != 0) {
			ucprcom_unlock(&(pShrm->ctrl.lock));
			shmdt(pShrm);
			ucprcom_sleep(3000);
			continue;
		}
		ucprcom_unlock(&(pShrm->ctrl.lock));

		while(1) {
			if(running_status != ucprcom_cli_status) {
				ucprcom_p("ucprcom: cli status [%d -> %d]\n", running_status, ucprcom_cli_status);
				ucprcom_lock(&(pShrm->ctrl.lock));
				if(ucprcom_cli_status <= 0) {
					ret = ucprcom_add_cli(pShrm, pid, 0);
				} else {
					ret = ucprcom_add_cli(pShrm, pid, ucprcom_cli_flag);
				}
				ucprcom_unlock(&(pShrm->ctrl.lock));
				if(ret == 0) {
					running_status = ucprcom_cli_status;
				} else {
					ucprcom_sleep(3000);
					continue;
				}
			}

			suspend = 0;
			if(running_status > 0 && ucprcom_cli_flag > 0) {
				if((ucprcom_cli_flag & UCPRCOM_FLAG_YUV) > 0 && ucprcom_cli_get_v(pShrm, &index_v, &timestamp_v) > 0) {
					suspend++;
				}
				if((ucprcom_cli_flag & UCPRCOM_FLAG_PCM) > 0 && ucprcom_cli_get_a(pShrm, &index_a, &timestamp_a) > 0) {
					suspend++;
				}
				if((ucprcom_cli_flag & UCPRCOM_FLAG_TS) > 0 && ucprcom_cli_get_ts(pShrm, &index_ts, &timestamp_ts) > 0) {
					suspend++;
				}
				if((ucprcom_cli_flag & UCPRCOM_FLAG_PIC) > 0 && ucprcom_cli_get_bmp(pShrm, &index_bmp, &timestamp_bmp) > 0) {
					suspend++;
				}
			}

			if(suspend <= 0) {
				if(fifo_fd < 0) {
					fifo_fd = open(fifo_file, O_RDONLY);
					if(fifo_fd >= 0) {
						ucprcom_p("ucprcom: open fifo OK\n");
					}
				}
				if(fifo_fd < 0) {
					ucprcom_p("ucprcom: open fifo error[%d:%s]\n", errno, strerror(errno));
					ucprcom_sleep(10);
				} else {
					if(read(fifo_fd, fifo_buf, sizeof(fifo_buf)) <= 0) {
						close(fifo_fd);
						fifo_fd = -1;
						ucprcom_p("ucprcom: read fifo error[%d:%s]\n", errno, strerror(errno));
						ucprcom_sleep(10);
					}
				}
			}
		}
	}

	ucprcom_p("ucprcom thread exit\n");

	return NULL;
}

int
ucprcom_cli_creat()
{
	pthread_attr_t attr;
	pthread_t thread;

	uc_p("ucprcom client version[%s]\n", UCPRCOM_VERSION);

	pthread_attr_init(&attr);
	pthread_attr_setdetachstate(&attr, PTHREAD_CREATE_DETACHED);
	if(pthread_create(&thread, &attr, ucprcom_cli_thread, NULL) != 0) {
		ucprcom_p("ucprcom: creat cli thread error[%d:%s]\n", errno, strerror(errno));
		pthread_attr_destroy(&attr);
		return -1;
	}
	pthread_attr_destroy(&attr);

	return 0;
}

void
ucprcom_srv_notify_cli(ucprcom_main_s *pUCPr, int type, ucprcom_timestamp_s *creat_time)
{
	char fifo_file[512];
	int i = 0;

	for(i = 0; i < UCPRCOM_SHRM_CLI_MAX; i++) {
		if(pUCPr->pShrm->ctrl.arr_pid[i] <= 0) {
			if(pUCPr->arr_fifo_pid[i] != 0) {
				if(pUCPr->arr_fifo_fd[i] >= 0) {
					close(pUCPr->arr_fifo_fd[i]);
					pUCPr->arr_fifo_fd[i] = -1;
				}
				pUCPr->arr_fifo_pid[i] = 0;
			}
			continue;
		}
		if(ucprcom_get_bit(pUCPr->pShrm->ctrl.arr_flag[i], type) > 0
			|| (pUCPr->arr_fifo_pid[i] <= 0 && pUCPr->arr_fifo_pid[i] != pUCPr->pShrm->ctrl.arr_pid[i])
			|| (pUCPr->arr_fifo_pid[i] == pUCPr->pShrm->ctrl.arr_pid[i]
				&& ucprcom_uptime_diff_ms(&pUCPr->arr_fifo_time[i], creat_time) >= 3000)) {
			if(kill(pUCPr->pShrm->ctrl.arr_pid[i], 0) != 0) {
				ucprcom_p("ucprcom: [%d] pid[%d] not exist\n", i, pUCPr->pShrm->ctrl.arr_pid[i]);
				ucprcom_del_cli_by_index(pUCPr->pShrm, i);
				if(pUCPr->arr_fifo_fd[i] >= 0) {
					close(pUCPr->arr_fifo_fd[i]);
					pUCPr->arr_fifo_fd[i] = -1;
				}
				pUCPr->arr_fifo_pid[i] = 0;
				continue;
			}
			if(pUCPr->arr_fifo_pid[i] != pUCPr->pShrm->ctrl.arr_pid[i]) {
				if(pUCPr->arr_fifo_fd[i] >= 0) {
					close(pUCPr->arr_fifo_fd[i]);
					pUCPr->arr_fifo_fd[i] = -1;
				}
				sprintf(fifo_file, "%sucprcom_%d", UCPRCOM_FIFO_PATH, pUCPr->pShrm->ctrl.arr_pid[i]);
			}
			if(pUCPr->arr_fifo_fd[i] < 0) {
				pUCPr->arr_fifo_fd[i] = open(fifo_file, O_WRONLY | O_NONBLOCK);
				if(pUCPr->arr_fifo_fd[i] >= 0) {
					pUCPr->arr_fifo_pid[i] = pUCPr->pShrm->ctrl.arr_pid[i];
					memcpy(&pUCPr->arr_fifo_time[i], creat_time, sizeof(ucprcom_timestamp_s));
					ucprcom_p("ucprcom: [%d] pid[%d] open fifo[%d] OK\n", i, pUCPr->pShrm->ctrl.arr_pid[i], pUCPr->arr_fifo_fd[i]);
				}
			}
			if(pUCPr->arr_fifo_fd[i] >= 0) {
				memcpy(&pUCPr->arr_fifo_time[i], creat_time, sizeof(ucprcom_timestamp_s));
				if(write(pUCPr->arr_fifo_fd[i], "a", 1) <= 0) {
					ucprcom_p("ucprcom: [%d] pid[%d] write fifo[%d] error[%d:%s]\n", i, pUCPr->pShrm->ctrl.arr_pid[i], pUCPr->arr_fifo_fd[i], errno, strerror(errno));
					close(pUCPr->arr_fifo_fd[i]);
					pUCPr->arr_fifo_fd[i] = -1;
				}
			}
		}
	}
}

void *
ucprcom_srv_thread(void *p)
{
	ucprcom_main_s *pUCPr = (ucprcom_main_s *)p;
	ucprcom_data_head_s *pHead = NULL;
	UC_U8 *pData = NULL;
	ucprcom_timestamp_s creat_time;
	int i = 0;
	int index = 0;

	pUCPr->if_thread = 1;

	while(pUCPr->if_run) {
		index = pUCPr->index_v;
		for(i = 0; i < UCPRCOM_SHRM_CACHE; i++) {
			index++;
			if(index >= UCPRCOM_SHRM_CACHE) {
				index = 0;
			}
			if(pUCPr->flag_v[index] > 0) {
				pHead = &(pUCPr->pShrm->arr_v[index].head);
				pData = pUCPr->pShrm->arr_v[index].data;
				ucprcom_lock(&(pHead->lock));
				if(ucprcom_get_bit(pHead->flag, UCPRCOM_DATA_FALG_READY) > 0
					&& ucprcom_get_bit(pHead->flag, UCPRCOM_DATA_FALG_OUTPUT) <= 0) {
					pUCPr->pFunc(pHead->type, &(pHead->fmt), pData, pHead->len, pUCPr->p);
					ucprcom_set_bit(pHead->flag, UCPRCOM_DATA_FALG_OUTPUT);
				}
				ucprcom_unlock(&(pHead->lock));
				pUCPr->flag_v[index] = 0;
				ucprcom_uptime(&creat_time, NULL, 0);
				ucprcom_srv_notify_cli(pUCPr, UCPRCOM_DATA_TYPE_YUV, &creat_time);
			}
		}

		ucprcom_sig_wait_timeout(&pUCPr->thread_sig, 0);
	}

	ucprcom_p("ucprcom server thread exit\n");

	pUCPr->if_thread = 0;

	return NULL;
}

void *
ucprcom_srv_creat(ucprcom_func pFunc, void *p)
{
	ucprcom_main_s *pUCPr = NULL;
	ucprcom_timestamp_s creat_time;
	pthread_attr_t attr;
	pthread_t thread;
	char buf[128];
	int i = 0;

	uc_p("ucprcom server version[%s]\n", UCPRCOM_VERSION);

	pUCPr = (ucprcom_main_s *)malloc(sizeof(ucprcom_main_s));
	if(pUCPr == NULL) {
		ucprcom_p("ucprcom: creat thread error[%d:%s]\n", errno, strerror(errno));
		return NULL;
	}
	ucprcom_uptime(&creat_time, NULL, 0);
	memset(pUCPr, 0, sizeof(ucprcom_main_s));
	for(i = 0; i < UCPRCOM_SHRM_CLI_MAX; i++) {
		pUCPr->arr_fifo_fd[i] = -1;
		memcpy(&pUCPr->arr_fifo_time[i], &creat_time, sizeof(ucprcom_timestamp_s));
	}

	ucprcom_system_ret(buf, sizeof(buf), "ipcs -m | grep 0x%08x | awk '{if($2~/^[0-9]+$/)print $2}'", UCPRCOM_SHRM_KEY);
	ucprcom_trim_to(buf, buf, sizeof(buf));
	if(strlen(buf) > 0) {
		ucprcom_p("ucprcom: shmid[%s] exist\n", buf);
	}

	pUCPr->shmid = shmget(UCPRCOM_SHRM_KEY, sizeof(ucprcom_shrm_s), IPC_CREAT | 0666);
	if (pUCPr->shmid < 0) {
		ucprcom_p("ucprcom: creat shrm error[%d:%s] del id[%s]\n", errno, strerror(errno), buf);
		if(strlen(buf) > 0) {
			ucprcom_system("ipcrm shm %s", buf);
		}
		return NULL;
	}

	pUCPr->pShrm = (ucprcom_shrm_s *)shmat(pUCPr->shmid, NULL, 0);
	if(pUCPr->pShrm == NULL) {
		ucprcom_p("ucprcom: shmat error[%d:%s]\n", errno, strerror(errno));
		return NULL;
	}

	ucprcom_sig_init(&pUCPr->thread_sig);
	pUCPr->if_run = 1;
	if(pFunc != NULL) {
		pUCPr->pFunc = pFunc;
		pUCPr->p = p;
		pthread_attr_init(&attr);
		pthread_attr_setdetachstate(&attr, PTHREAD_CREATE_DETACHED);
		if(pthread_create(&thread, &attr, ucprcom_srv_thread, pUCPr) != 0) {
			ucprcom_p("ucprcom: creat srv thread error[%d:%s]\n", errno, strerror(errno));
			pthread_attr_destroy(&attr);
			pUCPr->if_run = 0;
			return NULL;
		}
		pthread_attr_destroy(&attr);
	}

	if(pUCPr->pShrm->if_inited <= 0) {
		ucprcom_p("ucprcom: init shrm\n");
		ucprcom_lock_init(&(pUCPr->pShrm->ctrl.lock));
		for(i = 0; i < UCPRCOM_SHRM_CLI_MAX; i++) {
			pUCPr->pShrm->ctrl.arr_flag[i] = 0;
			pUCPr->pShrm->ctrl.arr_pid[i] = 0;
		}

		for(i = 0; i < UCPRCOM_SHRM_CACHE; i++) {
			ucprcom_lock_init(&(pUCPr->pShrm->arr_v[i].head.lock));
			pUCPr->pShrm->arr_v[i].head.flag = 0;
			ucprcom_lock_init(&(pUCPr->pShrm->arr_a[i].head.lock));
			pUCPr->pShrm->arr_a[i].head.flag = 0;
			ucprcom_lock_init(&(pUCPr->pShrm->arr_ts[i].head.lock));
			pUCPr->pShrm->arr_ts[i].head.flag = 0;
			ucprcom_lock_init(&(pUCPr->pShrm->arr_pic[i].head.lock));
			pUCPr->pShrm->arr_pic[i].head.flag = 0;
		}

		pUCPr->pShrm->if_inited = 1;
	} else {
		ucprcom_p("ucprcom: reconnect shrm\n");
	}

	ucprcom_lock(&(pUCPr->pShrm->ctrl.lock));
	pUCPr->pShrm->ctrl.index_v = 0;
	pUCPr->pShrm->ctrl.index_a = 0;
	pUCPr->pShrm->ctrl.index_ts = 0;
	pUCPr->pShrm->ctrl.index_bmp = 0;

	for(i = 0; i < UCPRCOM_SHRM_CLI_MAX; i++) {
		if(kill(pUCPr->pShrm->ctrl.arr_pid[i], 0) != 0) {
			ucprcom_p("ucprcom: [%d] pid[%d] not exist\n", i, pUCPr->pShrm->ctrl.arr_pid[i]);
			ucprcom_del_cli_by_index(pUCPr->pShrm, i);
		}
	}
	ucprcom_unlock(&(pUCPr->pShrm->ctrl.lock));

	ucprcom_lock(&(pUCPr->pShrm->arr_v[0].head.lock));
	pUCPr->pShrm->arr_v[0].head.flag = 0;
	ucprcom_unlock(&(pUCPr->pShrm->arr_v[0].head.lock));

	ucprcom_lock(&(pUCPr->pShrm->arr_a[0].head.lock));
	pUCPr->pShrm->arr_a[0].head.flag = 0;
	ucprcom_unlock(&(pUCPr->pShrm->arr_a[0].head.lock));

	ucprcom_lock(&(pUCPr->pShrm->arr_ts[0].head.lock));
	pUCPr->pShrm->arr_ts[0].head.flag = 0;
	ucprcom_unlock(&(pUCPr->pShrm->arr_ts[0].head.lock));

	ucprcom_lock(&(pUCPr->pShrm->arr_pic[0].head.lock));
	pUCPr->pShrm->arr_pic[0].head.flag = 0;
	ucprcom_unlock(&(pUCPr->pShrm->arr_pic[0].head.lock));

	return (void *)pUCPr;
}

int
ucprcom_srv_getp(void *pUCPrcom, int type, ucprcom_av_fmt_s **ppFmt, uint8_t **ppData, int *pDataMaxLen)
{
	ucprcom_main_s *pUCPr = (ucprcom_main_s *)pUCPrcom;
	ucprcom_data_head_s *pHead = NULL;
	UC_U8 *pData = NULL;

	if(pUCPrcom == NULL || ppFmt == NULL || ppData == NULL || pDataMaxLen == NULL || type < 0 || type >= UCPRCOM_DATA_TYPE_MAX) {
		ucprcom_p("ucprcom: getp entry error[%p, %d - %d, %p, %p, %p]\n", pUCPrcom, type, UCPRCOM_DATA_TYPE_MAX, ppFmt, ppData, pDataMaxLen);
		return -1;
	}

	*ppFmt = NULL;
	*ppData = NULL;
	*pDataMaxLen = 0;

	if(pUCPr->pShrm == NULL) {
		ucprcom_p("ucprcom: pShrm NULL\n");
		return -1;
	}

	switch(type) {
		case UCPRCOM_DATA_TYPE_YUV:
			pHead = &(pUCPr->pShrm->arr_v[pUCPr->index_v].head);
			pData = pUCPr->pShrm->arr_v[pUCPr->index_v].data;
			*pDataMaxLen = UCPRCOM_SHRM_V_MAX;
		break;
		case UCPRCOM_DATA_TYPE_PCM:
			pHead = &(pUCPr->pShrm->arr_a[pUCPr->index_a].head);
			pData = pUCPr->pShrm->arr_a[pUCPr->index_a].data;
			*pDataMaxLen = UCPRCOM_SHRM_A_MAX;
		break;
		case UCPRCOM_DATA_TYPE_TS:
			pHead = &(pUCPr->pShrm->arr_ts[pUCPr->index_ts].head);
			pData = pUCPr->pShrm->arr_ts[pUCPr->index_ts].data;
			*pDataMaxLen = UCPRCOM_SHRM_TS_MAX;
		break;
		case UCPRCOM_DATA_TYPE_PIC:
			pHead = &(pUCPr->pShrm->arr_pic[pUCPr->index_bmp].head);
			pData = pUCPr->pShrm->arr_pic[pUCPr->index_bmp].data;
			*pDataMaxLen = UCPRCOM_SHRM_PIC_MAX;
		break;
		default:
			ucprcom_p("ucprcom: type[%d] error\n", type);
			return -1;
		break;
	}

	if(ucprcom_lock(&(pHead->lock)) != 0) {
		ucprcom_p("ucprcom: lock error[%d:%s]\n", errno, strerror(errno));
		return -1;
	}

	pHead->type = type;
	pHead->flag = 0;
	pHead->len = 0;
	*ppFmt = &(pHead->fmt);
	*ppData = pData;

	return 0;
}

int
ucprcom_srv_putp(void *pUCPrcom, int type, int if_ok, int data_len, int if_outpout)
{
	ucprcom_main_s *pUCPr = (ucprcom_main_s *)pUCPrcom;
	ucprcom_data_head_s *pHead = NULL;
	UC_U32 *pIndexShrm = NULL;
	UC_U32 *pIndex = NULL;
	UC_U32 *pFlagV = NULL;
	ucprcom_timestamp_s creat_time;

	if(pUCPrcom == NULL || type < 0 || type >= UCPRCOM_DATA_TYPE_MAX) {
		ucprcom_p("ucprcom: putp entry error[%p, %d]\n", pUCPrcom, type);
		return -1;
	}
	if(pUCPr->pShrm == NULL) {
		ucprcom_p("ucprcom: pShrm NULL\n");
		return -1;
	}

	switch(type) {
		case UCPRCOM_DATA_TYPE_YUV:
			pHead = &(pUCPr->pShrm->arr_v[pUCPr->index_v].head);
			pIndex = &pUCPr->index_v;
			pIndexShrm = &pUCPr->pShrm->ctrl.index_v;
			pFlagV = &pUCPr->flag_v[pUCPr->index_v];
		break;
		case UCPRCOM_DATA_TYPE_PCM:
			pHead = &(pUCPr->pShrm->arr_a[pUCPr->index_a].head);
			pIndex = &pUCPr->index_a;
			pIndexShrm = &pUCPr->pShrm->ctrl.index_a;
		break;
		case UCPRCOM_DATA_TYPE_TS:
			pHead = &(pUCPr->pShrm->arr_ts[pUCPr->index_ts].head);
			pIndex = &pUCPr->index_ts;
			pIndexShrm = &pUCPr->pShrm->ctrl.index_ts;
		break;
		case UCPRCOM_DATA_TYPE_PIC:
			pHead = &(pUCPr->pShrm->arr_pic[pUCPr->index_bmp].head);
			pIndex = &pUCPr->index_bmp;
			pIndexShrm = &pUCPr->pShrm->ctrl.index_bmp;
		break;
		default:
			ucprcom_p("ucprcom: type[%d] error\n", type);
			return -1;
		break;
	}

	ucprcom_uptime(&creat_time, NULL, 0);
	if(if_ok > 0) {
		if(if_outpout > 0 || pUCPr->if_thread <= 0) {
			ucprcom_set_bit(pHead->flag, UCPRCOM_DATA_FALG_OUTPUT);
		} else {
			ucprcom_set_bit(pHead->flag, UCPRCOM_DATA_FALG_READY);
		}
		pHead->len = data_len;
		memcpy(&pHead->creat_time, &creat_time, sizeof(ucprcom_timestamp_s));
	}

	ucprcom_unlock(&(pHead->lock));

	if(if_ok > 0) {
		if(ucprcom_lock(&(pUCPr->pShrm->ctrl.lock)) != 0) {
			ucprcom_p("ucprcom: lock ctrl error[%d:%s]\n", errno, strerror(errno));
			return -1;
		}

		*pIndexShrm = *pIndex;
		if(pFlagV != NULL) {
			*pFlagV = 1;
		}

		*pIndex = *pIndex + 1;
		if(*pIndex >= UCPRCOM_SHRM_CACHE) {
			*pIndex = 0;
		}

		if(if_outpout > 0 || pUCPr->if_thread <= 0) {
			ucprcom_srv_notify_cli(pUCPr, type, &creat_time);
		} else {
			ucprcom_sig_signal(&pUCPr->thread_sig);
		}

		ucprcom_unlock(&(pUCPr->pShrm->ctrl.lock));
	}

	//ucprcom_p("index [%d][%d][%d]\n", pUCPr->pShrm->ctrl.index_v, pUCPr->pShrm->ctrl.index_a, pUCPr->pShrm->ctrl.index_ts);

	return 0;
}

void
ucprcom_srv_free(void *pUCPrcom)
{
	ucprcom_main_s *pUCPr = (ucprcom_main_s *)pUCPrcom;

	if(pUCPrcom == NULL) {
		ucprcom_p("ucprcom: free null\n");
		return;
	}

	pUCPr->if_run = 0;
	ucprcom_sig_signal(&pUCPr->thread_sig);
	ucprcom_sleep(1000);
	free(pUCPrcom);
}

#ifdef __cplusplus
}
#endif

